package com.sun.kafka.config;

import com.sun.kafka.service.TopicService;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.kafka.clients.admin.*;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.util.*;
import java.util.concurrent.ExecutionException;

/**
 * Topic初始化
 */
@Slf4j
@Component
@RequiredArgsConstructor(onConstructor = @__(@Autowired))
public class TopicInit implements InitializingBean {

    private final TopicService topicService;

    public String topics;

    @PostConstruct
    public void init() {
        List<String> list = topicService.listTopics();
        this.topics = StringUtils.join(list, ",");
    }

    @Override
    public void afterPropertiesSet() {
        // 初始化所有topic
        AdminClient adminClient = null;
        try {
            Properties properties = new Properties();
            properties.setProperty(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
            adminClient = AdminClient.create(properties);

            ListTopicsResult listTopicsResult = adminClient.listTopics();
            Set<String> names = listTopicsResult.names().get();

            List<String> allTopics = Arrays.asList("simple1", "simple2", "batch1", "batch2");
            List<NewTopic> newTopics = new ArrayList<>();
            for (String topic : allTopics) {
                if(!names.contains(topic)) {
                    log.info("初始化创建topic = {}", topic);
                    NewTopic newTopic = new NewTopic(topic, 2, (short) 1);
                    newTopics.add(newTopic);
                }
            }
            if(CollectionUtils.isNotEmpty(newTopics)) {
                adminClient.createTopics(newTopics);
            }
        } catch (InterruptedException e) {
            log.info("InterruptedException", e);
        } catch (ExecutionException e) {
            log.info("ExecutionException", e);
        } finally {
            if(adminClient != null) {
                adminClient.close();
            }
        }

        // 动态变量-可放入数据库中进行配置管理
        System.setProperty("monitor.kafka.topics", topics);
    }

}
